/**
 * 
 */
package jvs.peer.io;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;

import jvs.peer.PresenceListener;
import jvs.peer.io.Session.Callback;
import jvs.peer.misc.Base64;
import jvs.peer.sec.AccessControl;

/**
 * @author qiangli
 * 
 */
public abstract class DataPacketService implements MessageDriver, Messenger,
		Notifier, PeerStatus, FileTransfer, Lifecycle {
	protected static final String UTF8 = "UTF-8";

	private static final byte SYN = 0x16;
	private static final byte SOH = 0x01;
	private static final byte LF = 0x0A;
	
	private static final String TAB = "\t";
	
	protected static final byte TYPE_HEARTBEAT = 'H';
	protected static final byte TYPE_PACKET = 'P';

	protected static ArrayBlockingQueue<Message> collector = new ArrayBlockingQueue<Message>(
			Short.MAX_VALUE);
	
	protected boolean encode = false;

	public void init(Properties props) throws Exception {
	}

	public void release() {
	}

	public Monitor send(String to, String filename, File file) throws Exception {
		return null;
	}

	public Monitor receive(String from, String filename, File saveas)
			throws Exception {
		return null;
	}

	
	public String[] getNetAddress(String peer) {
		return null;
	}

	
	public void changeStatus(Status status) {
	}

	
	public Status checkStatus(String peer) {
		return null;
	}

	
	public void register(String peer, String tid, MessageListener ml) {
	}

	
	public void unregister(MessageListener ml) {

	}

	
	public void register(PresenceListener pl) {
	}

	
	public void unregister(PresenceListener pl) {
	}

	
	public void send(String peer, String tid, String message) throws Exception {
	}

	
	public Session createSession(String peer, Callback cb) throws Exception {
		return null;
	}

	
	public void invite(String peer) {
	}

	protected void send(Message msg, OutputStream os) throws IOException {
		final Header h = msg.getHeader();

		final String peer = h.getPeer();
		final String tid = h.getThread();
		final int mode = h.getMode();
		final String from = h.getFrom();
		//
		final int ioscmd = msg.getIoscmd();
		final String seqno = String.valueOf(msg.getSeqno());

		//
		if (encode) {
			encodeMessage(msg);
		}
		//
		final byte[] data = msg.getData();
		final int size = msg.getSize();

		// send
		BufferedOutputStream out = new BufferedOutputStream(os);
		try {
			out.write(SYN);
			out.write(SOH);
			out.write(TYPE_PACKET);
			//
			StringBuilder sb = new StringBuilder();
			// peer tid mode from ioscmd seqno size
			sb.append(peer);
			sb.append(TAB);
			sb.append(tid);
			sb.append(TAB);
			sb.append(mode);
			sb.append(TAB);
			sb.append(from);
			sb.append(TAB);
			sb.append(ioscmd);
			sb.append(TAB);
			sb.append(seqno);
			sb.append(TAB);
			sb.append(size);
			out.write(sb.toString().getBytes(UTF8));
			//	
			out.write(LF);
			
			//data
			if (data != null && size > 0) {
				out.write(data, 0, size);
			}
			
			//
			out.flush();
		} finally {
			out.close();
		}
	}
	
	protected String encode(byte[] buf, int len, String charset) throws UnsupportedEncodingException {
		if (len <=0 ) {
			return null;
		}
		return Base64.encodeBytes(buf,0,len);
	}
	
	protected byte[] decode(String s, String charset) throws UnsupportedEncodingException {
		if (s == null || s.length() == 0) {
			return new byte[0];
		}
		byte[] buf = s.getBytes(charset);
		return Base64.decode(buf, 0, buf.length);
	}

	/**
	 * Encode data and set the size
	 * @param msg
	 */
	protected void encodeMessage(Message msg) {
		//do nothing
	}
	
	/**
	 * Decode data and set the size
	 * @param msg
	 */
	protected void decodeMessage(Message msg) {
		//do nothing
	}

	public void receive(InputStream is) throws IOException {
		BufferedInputStream in = new BufferedInputStream(is);
		try {
			byte[] buf = new byte[1024];
			int c = 0;
			while (true) {
				c = in.read();
				//
				if (c == -1)  {
					return;
				}
				if (c == SYN) {
					continue;
				}
				
				if (c != SOH) {
					throw new IOException("Invalid data packet");
				}
				
				final int type = in.read();

				//read header
				int hlen = 0;
				while((c = in.read()) != LF) {
					if (c == -1) {
						throw new IOException("End of stream");
					}
					buf[hlen++] = (byte) c;
				}
				
				if (type == TYPE_HEARTBEAT) {
					continue;
				}
				
				//
				final String head = new String(buf, 0, hlen, UTF8);	
				String[] ha = head.split(TAB);
				//peer tid mode from ioscmd seqno size

				if (type == TYPE_PACKET) {
					Header h = new Header();
					Message msg = new Message();
					msg.setHeader(h);

					//
					h.setThread(ha[1]);
					h.setMode(Integer.parseInt(ha[2]));

					final String from = ha[3];

					h.setPeer(from);

					msg.setIoscmd(Integer.parseInt(ha[4]));
					msg.setSeqno(Integer.parseInt(ha[5]));

					final int size = Integer.parseInt(ha[6]);
					if (size > 0) {
						byte[] mbuf = new byte[size];
						int read = 0;
						for (int i = 0; i < size; i += read) {
							read = in.read(mbuf, i, size - i);
							if (read == -1) {
								throw new IOException("data packet corrupted");
							}
						}
						
						msg.setData(mbuf);
						msg.setSize(size);
					} else {
						msg.setData(null);
						msg.setSize(0);
					}
					
					if (encode) {
						decodeMessage(msg);
					}
					
					onMessage(msg);
				}
				
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			in.close();
		}
	}

	protected void onMessage(Message msg) throws InterruptedException {
		collector.put(msg);
	}

	public Message receive() throws IOException {
		try {
			return collector.take();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public String getJid() {
		return null;
	}

	
	public AccessControl getAccessControl() {
		return null;
	}

	
	public boolean ping(String peer) {
		return false;
	}

	
	public boolean available() {
		return true;
	}

	public abstract void shutdown() throws Exception;

	public abstract void startup() throws Exception;
	
	public abstract void send(Message msg) throws Exception;
}
